package com.thrift.server.annotation;

import com.thrift.server.rpc.server.TServerZk;
import com.thrift.server.rpc.server.ThriftServer;
import lombok.extern.slf4j.Slf4j;
import org.apache.thrift.TMultiplexedProcessor;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;

import java.lang.reflect.Constructor;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author 何锦涛--CodeSleep
 * @since 2022/9/11 20:36
 */
@Slf4j
public class ServeThread extends Thread {

    private int port;
    private Map<String, Object> serviceMap;
    private static final Logger logger = LoggerFactory.getLogger(ServeThread.class);
    // 线程池最小线程数和最大线程数，设置默认值，防止为空
    @Value("${server.thrift.min-thread-pool}")
    private final Integer minThreadPool = 100;
    @Value("${server.thrift.max-thread-pool}")
    private final Integer maxThreadPool = 1000;
    // 定义一个线程池
    private final ExecutorService thread = Executors.newFixedThreadPool(8);
    private final TServerZk tServerZk = new TServerZk("Thrift-test", "192.168.xxx.x:2181, 192.168.xxx.x:2182, 192.168.xxx.x:2183");

    public ServeThread(int port, Map<String, Object> serviceMap) {
        this.port = port;
        this.serviceMap = serviceMap;
    }

    @Override
    public void run() {
        try {
            TMultiplexedProcessor processor = new TMultiplexedProcessor();
            TNonblockingServerSocket socket = new TNonblockingServerSocket(port);
            THsHaServer.Args arg = new THsHaServer.Args(socket).minWorkerThreads(minThreadPool).maxWorkerThreads(maxThreadPool);
            arg.protocolFactory(new TBinaryProtocol.Factory());
            serviceMap.forEach((serviceName, serviceObjectMap) -> {
                // 获取 serviceImpl
                Class<?> serviceImplClass = serviceMap.get(serviceName).getClass();
                // 获取 Iface
                Class<?>[] interfaces = serviceImplClass.getInterfaces();
                for (Class<?> facade : interfaces) {
                    String facadeName = facade.getName();
                    if (facadeName.contains("$Iface")) {
                        String processorName = facadeName.substring(0, facadeName.indexOf("$")) + "$Processor";
                        Class<?> processorClass = null;
                        try {
                            processorClass = Class.forName(processorName);
                            Constructor<?> processorConstructor = processorClass.getConstructors()[0];
                            TProcessor tProcessor = (TProcessor) processorConstructor.newInstance(serviceMap.get(serviceName));
                            processor.registerProcessor(serviceName, tProcessor);
                            // 这里假设对于每一个 Iface，只有一个实现类
                            break;
                        } catch (Exception e) {
                            throw new RuntimeException(e.toString());
                        }
                    }
                }
                // 注册 Server 节点
                ThriftServer thriftServer = new ThriftServer(port, serviceName, tServerZk);
            });
            arg.processor(processor);
            TServer server = new THsHaServer(arg);
            thread.submit(server::serve);
            logger.info("启动 thrift server; port:" + port);
            int count = 0;
            while (!server.isServing()) {
                logger.error("等待启动 thrift server，轮训次数：{}", ++count);
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    logger.error("等待启动 thrift server 轮训发生线程异常");
                }
            }
            logger.info("thrift server; port:" + port + " 已经启动成功");
        } catch (Exception e) {
            logger.error("Thrift服务端启动失败，端口{}：{},", port, e);
        }
    }
}
